package com.thread.example.config;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Service;

import java.time.Duration;
import java.time.LocalDateTime;
import java.util.concurrent.*;

/**
 * 异步执行器
 *
 * @author hongjian.he
 */
@Service("asyncExecutor")
@Slf4j
public class AsyncExecutor implements InitializingBean {

    /**
     * 异步处理队列
     */
    private ThreadPoolExecutor executor;
    /**
     * 异步带返回值线程
     */
    private ThreadPoolExecutor callExecutor;
    private int threadPoolSize = 20;

    @Override
    public void afterPropertiesSet() throws Exception {
        BlockingQueue<Runnable> queue = new ArrayBlockingQueue<>(10);
        executor = new ThreadPoolExecutor(threadPoolSize, threadPoolSize, 0L, TimeUnit.SECONDS, queue);
        executor.allowCoreThreadTimeOut(false);
        // 当并发达到threadPoolSize+8，自身阻塞，不会消耗AMQ消息
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());

        BlockingQueue<Runnable> callQueue = new ArrayBlockingQueue<>(10);
        callExecutor = new ThreadPoolExecutor(threadPoolSize, threadPoolSize, 0L, TimeUnit.SECONDS, callQueue);
        callExecutor.allowCoreThreadTimeOut(false);
        // 当并发达到当并发达到threadPoolSize+8，自身阻塞，不会消耗AMQ消息
        callExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());

    }

    /**
     * 执行异步任务
     *
     * @param run
     */
    public void execute(Runnable run) {
        LocalDateTime beginTime = LocalDateTime.now();

        log.info("异步任务 ,开始时间:{}", beginTime);
        executor.execute(run);
        Long opetime = Duration.between(beginTime, LocalDateTime.now()).toMillis();
        log.info("异步任务RequestId={},结束时间:{}", Thread.currentThread().getName(), opetime);
    }

    /**
     * 执行异步任务
     *
     * @param call
     */
    public <T> Future<T> submit(Callable<T> call) {
        return callExecutor.submit(call);
    }
}
